package com.hopu.shop.pro;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.pro.ProUserTop3;
import com.hopu.util.HBaseUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.Properties;


public class ProUserTop3Ana {

    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder()
                .config("spark.driver.allowMultipleContexts", "true")
                .config("spark.sql.crossJoin.enabled", "true")
                .appName("ProUserTop3Ana")
                .master("local").getOrCreate();
        JavaSparkContext context = new JavaSparkContext(spark.sparkContext());
        JavaRDD<String> rdd = context.textFile("D://word/user_session.log");
        JavaRDD<ProUserTop3> map = rdd.map(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String province = json.getString("province");
            String uid = json.getString("uid");
            return new ProUserTop3(province, uid);
        });

        //创建DataFrame
        Dataset<Row> df = spark.createDataFrame(map, ProUserTop3.class);

        //去重
        Dataset<Row> distinct = df.distinct();

        //区域的用户人数
        Dataset<Row> province = distinct.groupBy("province").count();

        //排序
        Dataset<Row> order = province.orderBy(province.col("count").desc());
        order.show();

        //前三
        Dataset<Row> limit = order.limit(3);

        //写入数据库
        //保存到MySQL
        Properties pro = new Properties();
        pro.setProperty("driver", "com.mysql.jdbc.Driver");
        pro.setProperty("user", "root");
        pro.setProperty("password", "123456");
        limit.write().mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.136.200:3306/data_ana?useUnicode=true&characterEncoding=UTF-8", "pro_users_top3", pro);

        //保存到HBase
//        List<Row> rows = limit.collectAsList();
//        String[] columns = {"province", "count"};
//        for (Row row: rows) {
//            String[] values = {row.get(0).toString(), row.get(1).toString()};
//            try {
//                HBaseUtils.putOneRowToHbase("pro_user_top3", row.getString(0), "info", columns, values);
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
    }
}
